package com.flink.timewindow.watermark;

import com.flink.timewindow.bean.WaterSensor;
import com.flink.timewindow.function.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class WatermarkOutOfOrdernessDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //获取数据源
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("10.90.100.102", 8888)
                //数据处理
                //切分转换
                .map(new WaterSensorMapFunction());

        //TODO 指定Watermark策略
          //定义watermark策略
        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
                //指定watermark生成：乱序的，有等待时间：等待三秒
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                //指定时间戳分配器，从数据中提取
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        //返回的时间戳需要毫秒，所以得乘1000
                        System.out.println("数据="+element+" , recordTs="+recordTimestamp);
                        return element.getTs()*1000L;
                    }
                });
        SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);


        //分组
        KeyedStream<WaterSensor, String> sensorKS = sensorDSwithWatermark.keyBy(value -> value.getId());

        //窗口分配器
        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS.window(
                                                         //TODO   使用事件时间语义的窗口
                                                                 TumblingEventTimeWindows.of(Time.seconds(10))
                                                                 );
        //窗口函数
        SingleOutputStreamOperator<String> process = sensorWS.process(
                // 泛型： 1：输入类型  2：输出类型  3：分组key类型  4：窗口类型
                new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {

                    //全局函数计算逻辑： 窗口出发时才会调用一次，统一计算窗口的所有数据
                    //s: 分组的key  context：上下文   elements：存的数据   out：采集器
                    @Override
                    public void process(String s,
                                        ProcessWindowFunction<WaterSensor, String, String, TimeWindow>.Context context,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        //上下文可以拿到window对象，还有其他东西：如测流输出
                        long count = elements.spliterator().estimateSize();
                        long windowStartTs = context.window().getStart();
                        long windowEndTs = context.window().getEnd();
                        String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");
                        String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");

                        out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());



                    }
                }
        );
        //输出
        process.print();
        //执行
        env.execute();

    }
}
